package com.jiangjiang.eshop.inventory.async;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.concurrent.ArrayBlockingQueue;

/**
 * @ClassName StockUpdateQueueImpl
 * @Description TODO
 * @Author jhw
 * @Date 2021/6/14 14:22
 * @Version 1.0
 **/

@Component
public class StockUpdateQueueImpl implements StockUpdateQueue {
    private static final Integer QUEUE_MAX_SIZE = 1000;
    private final ArrayBlockingQueue<StockUpdateMessage> queue = new ArrayBlockingQueue<>(QUEUE_MAX_SIZE);

    @Autowired
    private OfflineStorageManager offlineStorageManager;


    @Override
    public void put(StockUpdateMessage message) throws Exception {
        // 检查离线存储标识，是否触发离线存储
        // 如果触发离线存储，则后续所有消息直接存储到mysql中，不在放入队列
        // 消息存储mysql完成后，检查当前阻塞队列大小，如果当前队列已经清零，
        // 则启动一个后台线程，将mysql中消息 按批次写入到内存队列中
        if(offlineStorageManager.getOffline()){
            offlineStorageManager.store(message);
            if(queue.size() == 0){
                new OfflineResumeThread(offlineStorageManager,this).start();
            }
            return;
        }

        if(QUEUE_MAX_SIZE.equals(queue.size())){
            offlineStorageManager.store(message);
            offlineStorageManager.setOffline(Boolean.TRUE);
            return;
        }
        queue.put(message);
    }

    @Override
    public StockUpdateMessage take() throws Exception {
        return queue.take();
    }

    @Override
    public void putDirect(StockUpdateMessage message) throws Exception {
        queue.put(message);
    }
}
